/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.lidejia.jobflow.taskhandler.ftp;

import com.snda.lidejia.jobflow.exception.JobFlowException;
import com.snda.lidejia.jobflow.task.TaskData;
import com.snda.lidejia.jobflow.task.TaskHandler;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPFileFilter;

/**
 *
 * @author lidejia
 */
public class FtpFileFetcher extends TaskHandler<FtpFetchList, String> {

    @Override
    public TaskData<String> handler(TaskData<FtpFetchList> inputs) throws JobFlowException {
        TaskData<String> output = new TaskData<String>();
        final Pattern filterPattern = Pattern.compile(getParam("filter"));
        for (FtpFetchList input : inputs) {
            try {
                FTPClient client = createFtpClient(input);
                try {
                    for (FilePair pair : input) {
                        String srcPath = pair.getSrc();
                        String destPath = pair.getDest();
                        FTPFile srcFile = client.mlistFile(srcPath);
                        if (srcFile.isDirectory()) {
                            FTPFile[] files = client.listFiles(srcPath, new FTPFileFilter() {

                                public boolean accept(FTPFile ftpf) {
                                    Matcher matcher = filterPattern.matcher(ftpf.getName());
                                    return matcher.find();
                                }
                            });
                            for (FTPFile file : files) {
                                String realSrcPath = srcPath + "/" + file.getName();
                                String realDestPath = destPath + "/" + file.getName();
                                if (fetchFile(client, realSrcPath, realDestPath)) {
                                    output.addData(realDestPath);
                                    System.out.println(realDestPath);
                                }
                            }
                        } else {
                            if (fetchFile(client, srcPath, destPath)) {
                                output.addData(destPath);
                                System.out.println(destPath);
                            }
                        }
                    }
                } finally {
                    if (client.isConnected()) {
                        client.disconnect();
                    }
                }
            } catch (IOException ex) {
                Logger.getLogger(FtpFileFetcher.class.getName()).log(Level.SEVERE, null, ex);
            } finally {
            }

        }
        return output;
    }

    private boolean fetchFile(FTPClient client, String src, String dest) throws IOException {
        boolean success = false;
        File file = new File(dest);
        File dir = file.getParentFile();
        dir.mkdirs();
        BufferedOutputStream outStream = new BufferedOutputStream(new FileOutputStream(file));
        try {
            if (client.retrieveFile(src, outStream)) {
                success = true;
            }
        } finally {
            outStream.close();
        }
        return success;
    }

    private FTPClient createFtpClient(FtpFetchList input) throws IOException {
        FTPClient client = new FTPClient();
        client.setConnectTimeout(60000);
        client.setDataTimeout(60000);
        client.connect(input.getHost(), input.getPort());
        client.login(input.getUserName(), input.getPassword());
        client.setSoTimeout(60000);
        client.setFileType(FTP.BINARY_FILE_TYPE);
        return client;
    }
}
